package com.reins.spark.controller;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Properties;

@Controller
@RequestMapping("/api/spark")
public class SparkController {
    @Autowired
    SparkSession sparkSession;

    @GetMapping("/basic")
    public String basicTest() {
        ClassPathResource classPathResource = new ClassPathResource("/src/main/resources/basic.txt");
        String filePath = classPathResource.getPath();
        JavaRDD<String> file = sparkSession.read().textFile(filePath).cache().toJavaRDD();
        JavaRDD<String> words = file.flatMap((FlatMapFunction<String, String>) s -> Arrays.asList(s.split(" ")).iterator());
        JavaPairRDD<String, Integer> wordToCountMap = words.mapToPair((PairFunction<String, String, Integer>) s -> new Tuple2<>(s, 1));
        JavaPairRDD<String, Integer> wordCounts = wordToCountMap.reduceByKey((Function2<Integer, Integer, Integer>) Integer::sum);
        wordCounts.saveAsTextFile("./word_count");

        return "ok";
    }

    @GetMapping("/parquet")
    public String parquetTest() {
        Properties properties = new Properties();
        properties.setProperty("user", "root");
        properties.setProperty("password", "123456");
        Dataset<Row> dataset = sparkSession.read().jdbc("jdbc:mysql://localhost:3306/spark", "person", properties);
        dataset.show();
        dataset.coalesce(1).write().mode(SaveMode.Overwrite).option("header", true).parquet("./test.parquet");
        return "ok";
    }
}
